package com.darchrow.demo.threadpool;

import com.alibaba.fastjson.JSONObject;
import com.darchrow.demo.threadpool.bean.DemoTaskBean;
import com.darchrow.demo.threadpool.bean.TaskBean;
import com.darchrow.demo.threadpool.constant.TaskKeyConstant;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author mdl
 * @date 2019/11/20
 */
@Slf4j
@Component
public class AsyncTaskConsumer {

    private Map<String, ITaskConsumer> taskConsumerMap = new HashMap<>();

    private OneTaskConsumer[] oneTaskArray;

    @Autowired
    private DemoTask demoTask;

    @Value("${async.task.thread.size}")
    private int asyncTaskThreadSize;

    private ExecutorService executorService;

    private AtomicInteger atomicInteger = new AtomicInteger(0);

    @PostConstruct
    public void init() {
        // 此处可以用工厂模式实现
        taskConsumerMap.put(TaskKeyConstant.DEMO_KEY, demoTask);
        if (asyncTaskThreadSize > 0) {
            ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("thread-call-runner-%d").build();
            executorService = new ThreadPoolExecutor(asyncTaskThreadSize, asyncTaskThreadSize, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), namedThreadFactory);
            oneTaskArray = new OneTaskConsumer[asyncTaskThreadSize];
            for (int i = 0; i < asyncTaskThreadSize; i++) {
                OneTaskConsumer oneTaskConsumer = new OneTaskConsumer("AsyncTask-" + i);
                oneTaskArray[i] = oneTaskConsumer;
                executorService.submit(oneTaskArray[i]);
            }

            // 这里给一个appendTask的入口
            new Thread(new Runnable() {
                @Override
                public void run() {
                    log.info("异步任务消费启动");
                    while (true) {
                        // 从外部一个地方获取数据，比如cache,或db
                        int index = atomicInteger.getAndIncrement() % asyncTaskThreadSize;
                        oneTaskArray[index].appendTask("{'taskType':'task_key_demo', 'commondata1':'abc','domainProperty1':'123'}");
                    }
                }
            }).start();

        }


    }

    class OneTaskConsumer extends Thread {

        /**
         * 线程的队列，按业务需求存放不同类型，例如json，dto
         */
        private final Queue<String> queue = new LinkedBlockingDeque<>();

        public OneTaskConsumer(String threadName) {
            super(threadName);
        }

        public void appendTask(String taskString) {
            queue.add(taskString);
        }


        @Override
        public void run() {
            while (true) {
                try {
                    log.info("线程启动:" + Thread.currentThread().getName());
                    String taskString = queue.poll();
                    if (StringUtils.isEmpty(taskString)) {
                        Thread.sleep(2000);
                        continue;
                    }
                    JSONObject json = JSONObject.parseObject(taskString);
                    String taskType = json.getString("taskType");
                    // json中可设置type，根据type
                    ITaskConsumer iTaskConsumer = taskConsumerMap.get(taskType);
                    if (null == iTaskConsumer) {
                        // 无效的任务类型
                        log.info("无效的任务类型~");
                        Thread.sleep(2000);
                        continue;
                    }
                    TaskBean taskBean = iTaskConsumer.buildTaskInfo(json);
                    if (null == taskBean) {
                        // 构造数据失败
                        log.info("构造数据失败~");
                        Thread.sleep(2000);
                        continue;
                    }
                    iTaskConsumer.executeTask(taskBean);

                } catch (Exception e) {
                    e.printStackTrace();
                }

            }

        }
    }


}


